package javaNIO1.Selectors;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.LinkedList;
import java.util.List;


/**
 * Created with Intel IDEA.
 *
 * @Auther: gjt
 * @Date: 2024/7/28 10:22
 */
public class SelectorSocketsThreadPool extends UseSelector {
    private final static int MAX_THREADS = 5;
    // 初始化线程池 并启动线程
    private static ThreadPool pool = new ThreadPool(MAX_THREADS);
    public static void main(String[] args) {
            new SelectorSocketsThreadPool().go(args);
    }

    protected  void readDataFromSocket(SelectionKey key) throws IOException {
        WorkThread worker = pool.getWorker();
        if(worker == null){
            return;
        }
        worker.serviceChannel(key);
    }

    private static class ThreadPool {
        List linkedList = new LinkedList();

        // 创建线程 并启动，并加入工作队列
        ThreadPool(int poolSize) {
            for (int i = 0; i < poolSize; i++) {
                WorkThread workThread = new WorkThread(this);
                workThread.setName("Worker" + (i+1));
                workThread.start();

                linkedList.add(workThread);
            }
        }

        // 获取第一个工作线程，并从工作队列中删除
        WorkThread getWorker() {
            WorkThread worker = null;
            synchronized (linkedList){
                if(linkedList.size()  > 0){
                    worker = (WorkThread) linkedList.remove(0);
                }
            }
            return worker;
        }

        // 回收线程 必须安全
        void returnWorker(WorkThread worker) {
            synchronized (linkedList){
                linkedList.add(worker);
            }
        }
    }


    private static class WorkThread extends Thread {
        private ByteBuffer buffer = ByteBuffer.allocateDirect(1024);
        private ThreadPool pool;
        private SelectionKey key;

        WorkThread(ThreadPool pool) {
            this.pool = pool;
        }

        // 必须得要同步
        public synchronized void run() {
            while (true) {
                try {
                    this.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    this.interrupted();
                }

                // 没有就绪的通道
                if(key == null)continue;

                System.out.println(this.getName() + "has been awakened");
                try{
                    drainChannel(key);
                }catch (Exception e){
                    System.out.println("Caught'" + e + "'closing channel");
                    try{
                        key.channel().close();
                    }catch (IOException ex){
                        ex.printStackTrace();
                    }
                    key.selector().wakeup();
                }
                key = null;
                this.pool.returnWorker(this);
            }
        }


        synchronized void serviceChannel(SelectionKey key){
            this.key = key;
            // ~SelectionKey.OP_READ 按位取反； 将感兴趣的操作从读取就绪状态移出 防止选择器重复的调用 readDataFromSocket
            key.interestOps(key.interestOps() & ~SelectionKey.OP_READ);
            // 唤醒线程
            this.notify();
        }

        void drainChannel(SelectionKey key) throws IOException {
            // 通过选择键获取 通道
            SocketChannel channel = (SocketChannel) key.channel();
            int count;

            // 清除缓冲区
            buffer.clear();
            // 读取通道中的数据
            while((count = channel.read(buffer)) > 0){
                buffer.flip();

                while(buffer.hasRemaining()){
                    channel.write(buffer);
                }
                buffer.clear();
            }
            // 读取完毕， 关闭通道
            if(count < 0){
                channel.close();
                return;
            }
            // key.interestOps() 获取当前的interest 集合（选择键感兴趣的事件的掩码）;
            // 改变interest集合中的感兴趣的操作，之前感兴趣的操作会保留，新增SelectionKey.OP_ACCEPT 操作(如果里边有此操作则不会改变)
            key.interestOps(key.interestOps() | SelectionKey.OP_ACCEPT);
            // 获取选择器并唤醒线程（唤醒select() 方法中休眠的线程）
            key.selector().wakeup();

        }
    }

}
